package com.sk.ccflower.server.common.tcpManager;

import java.nio.ByteBuffer;
import java.nio.ByteOrder;
import java.util.LinkedList;

import org.apache.log4j.Logger;

import com.sk.ccflower.lib.commonBeans.DefaultParamBean;
import com.sk.ccflower.lib.commonUtils.ObjectUtil;
import com.sk.ccflower.lib.commonUtils.StringUtil;
import com.sk.ccflower.lib.define.BaseCfg;
import com.sk.ccflower.lib.define.Consts;
import com.sk.ccflower.lib.meBeans.MessageBean;
import com.sk.ccflower.lib.meBeans.ScoreBean;
import com.sk.ccflower.lib.sockUtils.packet.DataHeader;
import com.sk.ccflower.lib.sockUtils.packet.MessageDefine;
import com.sk.ccflower.lib.sockUtils.packet.PacketDataUnit;
import com.sk.ccflower.server.lib.dbUtils.DBConnection;
import com.sk.ccflower.server.lib.dbUtils.qry.CfSpamListQry;
import com.sk.ccflower.server.lib.sysUtils.MessageQueueUtil;

public class InputDataHandler extends Thread {
	
	protected TcpManagerSession			m_parent;
	protected Logger					m_log;
	private int							nCurPos = 0;
	private int							nRecvLen = 0;
	private int							nBufferReadPos = 0;
	private int							nRecvBufferSize = 1000000;
	private int							nIncrease = 1000000;
	
	private ByteBuffer					readBuffer;
	private ByteBuffer					remainBuffer;
	private ByteBuffer					recvBuffer;
	
	private LinkedList<PacketDataUnit>	requestList;
	private boolean						runFlag;
	
	private DataHeader 					dataHeader = null;
	
	public InputDataHandler( TcpManagerSession _parent ) {
		m_parent = _parent;
		m_log = Logger.getLogger( getClass() );
		requestList = new LinkedList<PacketDataUnit>();
		readBuffer = ByteBuffer.allocate(4096);
		readBuffer.order(ByteOrder.BIG_ENDIAN);
		remainBuffer = ByteBuffer.allocate(4096);
		remainBuffer.order(ByteOrder.BIG_ENDIAN);
		
		recvBuffer = ByteBuffer.allocate(nRecvBufferSize);
		recvBuffer.order(ByteOrder.BIG_ENDIAN);
	}
	
	public int doRecvAction(byte[] b, int offset, int length) throws Exception {
		int nRead = -1;
		try {
			nRead = length;
			readBuffer.put( b, offset, length );
			//readBuffer.flip();
			readBuffer.rewind();
			
			// 처리된 데이터를 지운다.
			if ( nBufferReadPos >= (nRecvBufferSize / 10) )
			{
				ByteBuffer tmpBuffer = ByteBuffer.allocate(nRecvBufferSize);
				recvBuffer.rewind();
				recvBuffer.position(nBufferReadPos);
				tmpBuffer.put( recvBuffer.array(), 0, nCurPos );
				recvBuffer.clear();
				tmpBuffer.flip();
				recvBuffer.put( tmpBuffer.array(), 0, nCurPos );
				nRecvLen = nCurPos;
				nBufferReadPos = 0;
				
				m_log.info("buffer shift!!");
			}
			
			// 버퍼 크기가 모자라면 버퍼를 늘린다.
			if ( nRecvLen+nRead >= nRecvBufferSize )
			{
				ByteBuffer tmpBuffer = ByteBuffer.allocate(nRecvBufferSize);
				nRecvBufferSize += nIncrease;
				recvBuffer.flip();
				tmpBuffer.put( recvBuffer );
				tmpBuffer.flip();
				recvBuffer = null;
				recvBuffer = ByteBuffer.allocate(nRecvBufferSize);
				recvBuffer.put( tmpBuffer );
				
				m_log.info("buffer reallocate!!");
			}
			recvBuffer.rewind();
			recvBuffer.position( nRecvLen );
			recvBuffer.put( readBuffer.array(), 0, nRead );
			nRecvLen += nRead;
			nCurPos += nRead;
			
			while ( nCurPos > 0 )
			{
				if ( dataHeader == null ) {
					if ( nCurPos == DataHeader.HEADER_LENGTH ) {
						dataHeader = new DataHeader();
						byte[] header_buf = new byte[DataHeader.HEADER_LENGTH];
						recvBuffer.rewind();
						recvBuffer.position(nBufferReadPos);
						recvBuffer.get( header_buf, 0, DataHeader.HEADER_LENGTH );
						dataHeader.SetData( header_buf );
						
						nBufferReadPos += DataHeader.HEADER_LENGTH;
						nCurPos -= DataHeader.HEADER_LENGTH;
						
						if ( dataHeader.IsValidHeader() ) {
							PacketDataUnit pdu = new PacketDataUnit();
							pdu.CopyHeader(dataHeader);
							enqueue(pdu);
							
							dataHeader = null;
						} else {
							m_log.error( "invalid data header!!" );
							dataHeader = null;
						}
					} else if ( nCurPos > DataHeader.HEADER_LENGTH ) {
						dataHeader = new DataHeader();
						byte[] header_buf = new byte[DataHeader.HEADER_LENGTH];
						recvBuffer.rewind();
						recvBuffer.position(nBufferReadPos);
						recvBuffer.get( header_buf, 0, DataHeader.HEADER_LENGTH );
						dataHeader.SetData( header_buf );
						
						nBufferReadPos += DataHeader.HEADER_LENGTH;
						nCurPos -= DataHeader.HEADER_LENGTH;
					} else {
						break;
					}
				} else {
					if ( dataHeader.IsValidHeader() ) {
						if ( nCurPos >= dataHeader.len ) {
							int len = (int)dataHeader.len;
							ByteBuffer buffer = ByteBuffer.allocate(len);
							byte[] tmp_buf = new byte[len];
							recvBuffer.rewind();
							recvBuffer.position(nBufferReadPos);
							recvBuffer.get( tmp_buf, 0, len );
							buffer.put( tmp_buf, 0, len );
							
							nBufferReadPos += len;
							nCurPos -= len;
							
							PacketDataUnit pdu = new PacketDataUnit();
							pdu.CopyHeader(dataHeader);
							pdu.CopyBuffer(buffer);
							enqueue(pdu);
							
							dataHeader = null;
						} else {
							break;
						}
					} else {
						m_log.error( "invalid data header!!" );
						dataHeader = null;
					}
				}
			}
		} catch(Exception ex) {
			m_log.error( ex );
			ex.printStackTrace();
			throw ex;
		}
		return nRead;
	}
	
	public void run() {
		runFlag = true;
		PacketDataUnit pdu = null;
		
		try {
			while ( runFlag ) {
				pdu = dequeue();
				if ( pdu == null ) {
					Thread.sleep(10);
					continue;
				}
				
				m_log.debug( pdu.header.getHeaderLog() );
				
				if ( pdu.header.mid == MessageDefine.LOGIN ) {
					m_parent.m_email = new String( pdu.body.array() );
					
					/**
					 *  실시간 데이터 처리를 위해 접속세션을 TcpManager의 m_hashClient에 이메일을 키로 저장하기 위해 MQ로 전송
					 */
					DefaultParamBean bean = new DefaultParamBean();
					bean.src_user = m_parent.m_email;
					MessageQueueUtil.sendMsg( BaseCfg.MESSAGE_QUEUE_SERVER_URL, Consts.MESSAGE_QUEUE_NAME.TCP_MANAGER_CLIENTS, bean);
				}
				else if ( pdu.header.mid == MessageDefine.HEART_BEAT ) {
					sendHeartBeatReply( pdu.header );
				}
				else if ( pdu.header.mid == MessageDefine.FACE_SCORE ) {
					String strBody = new String( pdu.body.array(), 0, pdu.body.array().length, "UTF-8" );
					String[] values = StringUtil.toStringArray( strBody, String.valueOf(MessageDefine.SEP) );
					ScoreBean bean = new ScoreBean(values);
					
					// 캐시 처리
					
					
					MessageQueueUtil.sendMsg( BaseCfg.MESSAGE_QUEUE_SERVER_URL, Consts.MESSAGE_QUEUE_NAME.MESSENGER_FACE_SCORE, bean );
					// 상대방 TCP 세션에 실시간 전송
					MessageBean mBean = new MessageBean();
					ObjectUtil.objectCopy( mBean, bean, false );
					sendChatData( mBean );
				}
				else if ( pdu.header.mid == MessageDefine.CHAT ) {
					String strBody = new String( pdu.body.array(), 0, pdu.body.array().length, "UTF-8" );
					String[] values = StringUtil.toStringArray( strBody, String.valueOf(MessageDefine.SEP) );
					MessageBean bean = new MessageBean(values);
					
					// 차단 여부 체크
					DBConnection dbconn = new DBConnection(BaseCfg.JDBC_URL, BaseCfg.JDBC_USER, BaseCfg.JDBC_PWD);
					dbconn.connect();
					CfSpamListQry qry = new CfSpamListQry( dbconn );
					qry.selectBlockCheck( bean.src_user, bean.dest_user );
					if ( qry.m_arr.size() > 0 ) {
						m_log.debug( "대화상대 차단 : "+bean.dest_user+" -> "+bean.src_user );
						sendErrorChatBlocked();
						continue;
					}
					dbconn.close();
					
					// DB에 저장
					MessageQueueUtil.sendMsg( BaseCfg.MESSAGE_QUEUE_SERVER_URL, Consts.MESSAGE_QUEUE_NAME.MESSENGER_CHAT, bean );
					// 상대방 TCP 세션에 실시간 전송
					sendChatData( bean );
				}
				
				pdu = null;
			}
		} catch ( Exception ex ) {
			m_log.error( ex );
			ex.printStackTrace();
		}
	}
	
	public synchronized void enqueue( PacketDataUnit data ) {
		requestList.add(data);
	}
	
	private synchronized PacketDataUnit dequeue() {
		if ( requestList.size()==0 ) {
			return null;
		}
		PacketDataUnit data = (PacketDataUnit)requestList.get(0);
		requestList.remove(0);
		return data;
	}
	/*
	private int indexOf ( byte[] buffer, int s, int n, byte value ) {
		for ( int i=s; i<n; i++ ) {
			if ( buffer[i]==value )	return i;
		}
		return -1;
	}
	*/
	private int sendHeartBeatReply( DataHeader header ) throws Exception {
		header.type = 'P';
		header.ts = (int) (System.currentTimeMillis() / 1000);
		
		byte[] send_buf = header.GetByteBuffer().array();
		
		int ret = m_parent.send( send_buf );
		
		m_log.debug( "sendHeartBeatReply() : "+ret );
		m_log.debug( header.getHeaderLog() );
		
		return ret;
	}
	
	private int sendErrorChatBlocked() throws Exception {
		DataHeader header = new DataHeader();
		header.type	= 'I';
		header.mid = MessageDefine.ERROR_CHAT_BLOCKED;
    	header.len = 0;
		
		byte[] send_buf = header.GetByteBuffer().array();
		
		int ret = m_parent.send( send_buf );
		
		m_log.debug( "sendErrorChatBlocked() : "+ret );
		m_log.debug( header.getHeaderLog() );
		
		return ret;
	}
	
	private void sendChatData( MessageBean bean ) throws Exception {
		TcpManagerSession session = m_parent.findSession( bean.dest_user );
		if ( null != session ) {
			try {
				m_log.debug( "sendChatData() : found destination session!" );
				m_log.debug( "sendChatData() : Destination -> IP ("+session.getSockChannel().socket().getInetAddress().getHostAddress()+") Port("+session.getSockChannel().socket().getPort()+")" );
				
		    	String body = bean.src_user;
		    	body += MessageDefine.SEP;
		    	body += bean.dest_user;
		    	body += MessageDefine.SEP;
		    	body += bean.comment;
		    	byte[] body_buf = body.getBytes( "UTF-8" );
		    	
		    	DataHeader header = new DataHeader();
		    	header.mid = MessageDefine.CHAT;
		    	header.len = body_buf.length;
		    	byte[] haader_buf = header.GetByteBuffer().array();
		    	int len = haader_buf.length + body_buf.length;
		    	byte[] total_buf = new byte[len];
		    	System.arraycopy( haader_buf, 0, total_buf, 0, haader_buf.length);
		    	System.arraycopy( body_buf, 0, total_buf, haader_buf.length, body_buf.length);
		    	
		    	int ret = session.send( total_buf );
		    	m_log.debug( "sendChatData() : ret("+ret+")" );
			} catch (Exception e) {
				m_log.error( e );
				e.printStackTrace();
				if ( e.getMessage().contains("Broken") ) {
					
				}
			}
		} else {
			m_log.debug( "sendChatData() : not found destination session!" );
		}
	}
	
}
